package com.ccf.business.etl.service.impl;

import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.ccf.business.etl.model.dto.EdgeDTO;
import com.ccf.business.etl.model.dto.FlowInfoDTO;
import com.ccf.business.etl.model.dto.NodeDTO;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.component.api.constants.ComponentCodeConstants;
import com.component.api.constants.FlowConstants;
import com.component.api.constants.FlowStatus;
import com.component.api.constants.StatusContents;
import com.component.api.model.flow.FlowInfo;
import com.component.api.model.flow.NodeInfo;
import com.component.api.model.flow.RelationInfo;
import com.component.api.model.param.TableColumn;
import com.component.api.utils.FlowUtils;
import com.common.utils.constant.Constants;
import com.common.utils.tool.StringUtil;
import com.ccf.business.etl.flow.ParamManager;
import com.ccf.business.etl.flow.mq.FlowRunningProducer;
import com.ccf.business.etl.mapper.FlowMapper;
import com.ccf.business.etl.mapper.TaskMapper;
import com.ccf.business.etl.model.Flow;
import com.ccf.business.etl.model.Node;
import com.ccf.business.etl.model.Task;
import com.ccf.business.etl.service.IFlowService;
import com.ccf.business.etl.service.INodeService;
import com.common.model.exception.FlowAnalysisException;
import com.common.model.exception.FlowRunningException;
import com.common.model.pojo.Query;
import com.common.model.result.CodeEnum;
import com.common.model.result.PageResult;
import com.service.mybatis.service.impl.SuperFlagServiceImpl;
import com.service.mybatis.service.impl.SuperServiceImpl;
import com.service.mybatis.util.Condition;
import com.framework.redis.service.RedisService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * 流程表
 *
 * @author ccf
 * @date 2021-11-29 18:02:10
 */
@Slf4j
@Service
public class FlowServiceImpl extends SuperFlagServiceImpl<FlowMapper, Flow> implements IFlowService {

    @Autowired
    private INodeService nodeService;

    @Autowired
    private FlowRunningProducer flowRunningProducer;

    @Autowired
    private ParamManager paramManager;

    @Autowired
    private TaskMapper taskMapper;

    @Autowired
    private RedisService redisService;

    /**
    * 列表
    * @param query
    * @return
    */
    @Override
    public PageResult<Flow> findList(Query query){
        Page<Flow> page = page(Condition.getPage(query), Condition.getQueryWrapper(query));
        return PageResult.<Flow>builder().datas(page.getRecords()).code(CodeEnum.SUCCESS.getCode()).count(page.getTotal()).build();
    }

    @Override
    public String saveAndAnalysis(Flow flow) {
        String data = flow.getData();
        if (StrUtil.isBlank(data)){
            saveOrUpdate(flow);
            return flow.getId();
        }

        //流程id 提交生成
        String flowId = flow.getId();
        if (StrUtil.isEmpty(flow.getId())){
            flowId = IdUtil.fastSimpleUUID();
        }

        FlowInfoDTO flowInfoDTO = JSON.parseObject(data, FlowInfoDTO.class);
        flowInfoDTO.setName(flow.getName());
        flowInfoDTO.setRemark(flow.getRemark());
        flowInfoDTO.setId(flowId);

        //修改采用覆盖的方式，因为有可能有太多改动
        updateNodes(flowInfoDTO);

        //验证流程合法性
        checkFlowInfo(flowInfoDTO);

        flow.setData(JSON.toJSONString(flowInfoDTO));
        if (StrUtil.isEmpty(flow.getId())){
            flow.setId(flowId);
            save(flow);
            return flowId;
        } else {
            updateById(flow);
            return flow.getId();
        }
    }

    /**
     * 检测流程合法性
     * @param flowInfoDTO
     */
    private void checkFlowInfo(FlowInfoDTO flowInfoDTO) {
        List<NodeDTO> nodes = flowInfoDTO.getNodes();
        Set<String> ids = nodes.stream().map(e -> e.getId()).collect(Collectors.toSet());
        List<EdgeDTO> edges = flowInfoDTO.getEdges();
        for (int i = 0; i < edges.size() ; i++) {
            EdgeDTO edgeDTO = edges.get(i);
            String source = edgeDTO.getSource();
            String target = edgeDTO.getTarget();
            if (!(ids.contains(source) && ids.contains(target))){
                edges.remove(i);
                i--;
            }
        }
    }

    @Override
    public FlowInfoDTO getDetailById(String flowId) {
        Flow flow = getById(flowId);
        String data = flow.getData();
        if (StrUtil.isBlank(data)){
            return new FlowInfoDTO(flow.getId(), flow.getName(), flow.getRemark());
        }
        FlowInfoDTO flowInfoDTO = JSON.parseObject(data, FlowInfoDTO.class);
        Map<String, Object> paramMap = Maps.newHashMap();
        paramMap.put("flow_id", flowInfoDTO.getId());
        List<Node> nodes = nodeService.listByMap(paramMap);
        flowInfoDTO.setNodes(nodes.stream().map(e->new NodeDTO(e)).collect(Collectors.toList()));
        return flowInfoDTO;
    }

    /**
     * 保存节点
     * @param flowInfoDTO
     */
    private void updateNodes(FlowInfoDTO flowInfoDTO) {
        Map<String, Object> paramMap = Maps.newHashMap();
        paramMap.put("flow_id", flowInfoDTO.getId());
        nodeService.removeByMap(paramMap);
        List<NodeDTO> nodeDTOs = flowInfoDTO.getNodes();
        List<Node> nodes = new ArrayList<>();
        for (NodeDTO ndto : nodeDTOs) {
            //解析不能再这做 因为保存时不需要验证流程可执行性
            Node node = new Node();
            node.setFlowId(flowInfoDTO.getId());
            node.setName(ndto.getName());
            node.setRule(ndto.getCode());
            node.setId(ndto.getId());
            node.setIcon(ndto.getIcon());
            node.setEnterLine(ndto.getEnterLine());
            node.setOutputLine(ndto.getOutputLine());
            node.setParams(ndto.getParams());
            nodes.add(node);
            //参数信息单独保存，不然会导致数据太大太多太乱
            ndto.setParams(null);
        }
        if (nodes.size() > 0){
            nodeService.saveBatch(nodes);
        }
    }


    @Override
    public Flow testRunning(String id, String data) {

        if (StrUtil.isBlank(data)){
            throw new FlowAnalysisException("运行数据为空");
        }
        String flowId = FlowConstants.TEST_FLOW + IdUtil.fastSimpleUUID();
        FlowInfo flowInfo = dataToFlowInfo(data);
        flowInfo.setFlowId(flowId);
        flowInfo.setIsNeedViewResult(true);
        flowInfo.setFlowName("当前测试流程");
        //提交运行
        flowRunningProducer.running(flowInfo);
        //保存状态为提交
        redisService.set(StatusContents.FLOW_STATUS_KEY + flowId, FlowStatus.SUBMIT.toString());
        //记录运行
        if (StrUtil.isNotBlank(id)){
            Flow byId = getById(id);
            if (byId != null){
                byId.setJobId(flowId);
                saveOrUpdate(byId);
            }
        }
        return new Flow(flowId);
    }

    /**
     * 任务运行参照
     * @param flowId 运行
     * @return
     */
    @Override
    @Deprecated
    public String running(String flowId) {
        //两个必要参数 这里是获取任务实体
        Flow flow = getById(flowId);
        if (flow == null){
            throw new FlowRunningException("运行id不存在"+ flowId);
        }
        //这里是获取任务数据
        String data = flow.getData();
        String jobId =  flowId;
        FlowInfoDTO flowInfoDTO = JSON.parseObject(data, FlowInfoDTO.class);
        FlowInfo flowInfo = new FlowInfo();
        //这里改成任务id
        flowInfo.setFlowId(jobId);
        //这里改成任务名称
        flowInfo.setFlowName(flow.getName());
        flowInfo.setIsNeedViewResult(true);
        flowInfo.setEdges(flowInfoDTO.getEdges().stream()
                .map(e->
                        RelationInfo.builder().id(e.getId()).source(e.getSource()).target(e.getTarget()).build()
                ).collect(Collectors.toList()));
        List<NodeInfo> nodeInfos = Lists.newArrayList();
        Map<String, Object> paramMap = Maps.newHashMap();
        paramMap.put("flow_id", flowInfoDTO.getId());
        List<Node> nodes = nodeService.listByMap(paramMap);
        for(Node node : nodes){
            //任务执行时调用下面注释这个方法
//            nodeInfos.add(new NodeInfo(node.getId(),node.getName(), node.getRule(), node.getRunParams()));
            //任务执行不再解析节点
            try {
                NodeInfo nodeInfo = paramManager.getHandler(new NodeDTO(node)).get();
                nodeInfos.add(nodeInfo);
            } catch (Exception e) {
                e.printStackTrace();
                throw new FlowAnalysisException(node.getName() + "参数错误:"+e.getMessage());
            }
        }
        flowInfo.setNodes(nodeInfos);
        //提交运行
        flowRunningProducer.running(flowInfo);
        return jobId;
    }

    @Override
    public String copy(String flowId, String taskId) {
        Flow flow = getById(flowId);
        String data = flow.getData();
        if (StrUtil.isBlank(data)){
            throw new FlowAnalysisException("空流程无法运行");
        }
        Map<String, String> idMap = nodeService.copy(flowId, taskId);
        String[] oldIds = new String[idMap.size()];
        String[] newIds = new String[idMap.size()];
        int i = 0 ;
        for (Map.Entry<String, String> entry : idMap.entrySet()) {
            oldIds[i] = entry.getKey();
            newIds[i] = entry.getValue();
            i++;
        }
        String newData = StringUtil.replaceEach(data, oldIds, newIds);
        return newData;
    }

    @Override
    public Map<String, List<TableColumn>>  selectFields(String data, String currentId) {
        FlowInfo flowInfo = dataToFlowInfo(data, true);
        return FlowUtils.pretreatmentFields(flowInfo, currentId);
    }

    @Override
    public boolean deleteValidate(String id) {
        List<String> ids = StrUtil.split(id, ',');

        long count = taskMapper.selectCount(Wrappers.<Task>lambdaQuery().in(Task::getFlowId, ids));
        if(count > 0) {
            return false;
        }
        return true;
    }


    private FlowInfo dataToFlowInfo(String data) {
        return dataToFlowInfo(data, false);
    }

    /**
     * 有些时候需要跳过错误继续执行
     * @param data
     * @param isErrorSkip 是否跳过错误
     * @return
     */
    private FlowInfo dataToFlowInfo(String data, boolean isErrorSkip) {
        if (StrUtil.isBlank(data)){
            throw new FlowAnalysisException("运行数据为空");
        }
        FlowInfoDTO flowInfoDTO = JSON.parseObject(data, FlowInfoDTO.class);
        FlowInfo flowInfo = new FlowInfo();
        flowInfo.setEdges(flowInfoDTO.getEdges().stream()
                .map(e->
                        RelationInfo.builder().id(e.getId()).source(e.getSource()).target(e.getTarget()).build()
                ).collect(Collectors.toList()));
        List<NodeDTO> nodes = flowInfoDTO.getNodes();
        List<NodeInfo> nodeInfos = Lists.newArrayList();
        for (NodeDTO nodeDto : nodes) {
            try {
                NodeInfo nodeInfo = paramManager.getHandler(nodeDto).get();
                nodeInfos.add(nodeInfo);
            } catch (Exception e) {
                //输入节点不能跳过
                if (isErrorSkip && !ComponentCodeConstants.DATA_INPUT.equalsIgnoreCase(nodeDto.getCode())) {
                    NodeInfo nodeInfo = NodeInfo.builder().name(nodeDto.getName())
                            .id(nodeDto.getId()).rule(nodeDto.getCode()).build();
                    nodeInfos.add(nodeInfo);
                    continue;
                }
                e.printStackTrace();
                throw new FlowAnalysisException(nodeDto.getName() + "参数配置错误:"+e.getMessage());
            }
        }
        flowInfo.setNodes(nodeInfos);
        return flowInfo;
    }

    @Override
    public boolean status(String id) {
        String status = null;
        if (StrUtil.startWith(id, FlowConstants.TEST_FLOW)){
            status = redisService.get(StatusContents.FLOW_STATUS_KEY + id);
        } else {
            Flow byId = getById(id);
            if (byId == null || byId.getJobId() == null){
                return false;
            }
            status = redisService.get(StatusContents.FLOW_STATUS_KEY + byId.getJobId());
        }
        if (status == null) {
            return false;
        }

        switch (FlowStatus.valueOf(status)){
            case RUNNING:
            case WAIT:
            case START:
            case SUBMIT:
                return true;
            default:
                return false;
        }
    }

    @Override
    public boolean stopTest(String flowId) {
        if (StrUtil.startWith(flowId, FlowConstants.TEST_FLOW)){
            //保存状态为停止
            redisService.set(StatusContents.FLOW_STATUS_KEY + flowId, FlowStatus.STOP.toString());
        }  else{
            Flow byId = getById(flowId);
            if (byId == null || byId.getJobId() == null){
                return false;
            }
            //保存状态为停止
            redisService.set(StatusContents.FLOW_STATUS_KEY + byId.getJobId(), FlowStatus.STOP.toString());

        }

        return true;
    }
}
